/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ban.ycl.data.exchange.runner.proxy;

import cn.ac.iie.di.ban.data.exchange.runner.commons.protocol.ScopeEnum;
import cn.ac.iie.di.ban.data.exchange.runner.commons.protocol.proxy.DataProxyRequest;
import cn.ac.iie.di.ban.data.exchange.runner.commons.protocol.proxy.DataProxyResponse;
import io.atomix.group.messaging.MessageProducer;
import java.util.HashMap;
import java.util.concurrent.CompletableFuture;
import org.apache.log4j.Logger;

/**
 *
 * @author Administrator
 */
public class MasterTools {

    private static Logger logger = Logger.getLogger(ParserFastJson.class);

    public static synchronized DataProxyResponse sendToSlave(MessageProducer<DataProxyRequest> producer, HashMap dataMap) {
        try {
            String sql = (String) dataMap.get("sql");
            ScopeEnum scope = (ScopeEnum) dataMap.get("scope");
            String uniq_key = (String) dataMap.get("uniq_key");
            String uid = (String) dataMap.get("uid");
            DataProxyRequest taskOrder = new DataProxyRequest(uid, sql, scope, uniq_key);

//            AtomixClient atclient = AtomixClient.builder().withTransport(new NettyTransport()).build();
//            ProtocolTools.SerializeAll(atclient.serializer());
//            atclient.connect(clusterAddress).join();
//            DistributedGroup group = atclient.getGroup(groupName).join();
//            Election election = group.election();
//            Term term = election.term();
//            GroupMember leader = term.leader();
//            MessageClient messaging = null;
//            int trynum = 0;
//            while (trynum < 3) {
//                trynum++;
//                try {
//                    messaging = leader.messaging();
//                } catch (Exception e) {
//                    if (trynum == 3) {
//                        logger.error(e.getMessage(), e);
//                        return null;
//                    }
//                    Thread.sleep(500);
//                }
//            }
//
//            MessageProducer<DataProxyRequest> producer = messaging.producer(producerName);
            logger.info("begin to send master!");
            CompletableFuture<DataProxyResponse> send = producer.send(taskOrder);
            logger.info("producer.send");
//            CompletableFuture<DataProxyResponse> send = group.election().term().leader().messaging().producer(producerName).send(taskOrder);
            DataProxyResponse resopnse = send.join();
            logger.info("send.join");

//            atclient.close();
            return resopnse;
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
            return null;
        }
    }
}
